package taskservice

import (
	"context"
	"errors"
	"fmt"
	"time"
	"yunj/pkg/global"
	"yunj/pkg/yunj/app/consts"
	"yunj/pkg/yunj/app/enum/admin/noticeenum"
	"yunj/pkg/yunj/app/enum/admin/taskenum"
	"yunj/pkg/yunj/app/model"
	"yunj/pkg/yunj/app/service/admin/noticeservice"
	"yunj/pkg/yunj/core/redis"
	"yunj/pkg/yunj/util"

	"gorm.io/gorm"
)

// 任务处理者接口
type TaskHandlerInterface interface {
	SetTask(task *model.AdminTask) // 设置任务
	GetTask() *model.AdminTask     // 获取任务
	Handle() error                 // 处理任务
}

// 创建任务并立即执行
func CreateHandle(task *model.AdminTask) (err error) {
	if err = Create(task); err != nil {
		return
	}
	err = Handle(task.Id)
	return
}

// 创建任务
func Create(task *model.AdminTask) (err error) {
	if task == nil || task.Id <= 0 {
		err = fmt.Errorf("任务模型数据获取空！")
		return
	}
	errPrefix := fmt.Sprintf("创建任务 %s[%d] ", task.Title, task.Id)
	currTime := time.Now()
	// 判断任务能否执行
	var existsTask model.AdminTask
	if dbErr := global.MySQL.Where("id = ?", task.Id).First(&existsTask).Error; dbErr != nil && !errors.Is(dbErr, gorm.ErrRecordNotFound) {
		err = fmt.Errorf("%s数据查询异常！%v", errPrefix, dbErr)
		return
	}
	if existsTask.Id > 0 {
		return
	}
	// 创建任务
	task.CreatedAt = &currTime // GenerateNextStatusSnapTimeLineNode内有使用CreatedAt，所以此处赋值须置前
	task.UpdatedAt = &currTime
	nextTimeLineNode, _ := task.GenerateNextStatusSnapTimeLineNode(currTime, taskenum.STATUS_PENDING)
	task.SetSnapTimeLineNode(nextTimeLineNode)
	task.Status = taskenum.STATUS_PENDING
	if dbErr := global.MySQL.Create(&task).Error; dbErr != nil {
		err = fmt.Errorf("%s创建失败，原因：%v", errPrefix, dbErr)
	}
	return
}

// 处理任务
func Handle(taskIds ...uint64) (err error) {
	lock := redis.NewLock("admin.task.handle", 300)
	defer lock.Release()
	var ok bool
	ok, err = lock.Get()
	if err != nil {
		err = fmt.Errorf("后台任务处理锁获取失败！%v", err)
		return
	}
	if !ok {
		return
	}
	// 获取待处理任务
	tasks, err := getPendingTasks(taskIds...)
	if err != nil || len(tasks) == 0 {
		return
	}
	// 任务处理
	for _, task := range tasks {
		go HandleTask(task.Id)
	}
	return
}

// 取消执行
func CancelHandle(taskIds []uint64, reason string) (err error) {
	if len(taskIds) == 0 {
		return
	}
	errPrefix := "任务重新执行："
	// 获取等待执行和执行中的任务
	var tasks []*model.AdminTask
	if dbErr := global.MySQL.Where("id in ? and status in ?", taskIds, []any{taskenum.STATUS_PENDING, taskenum.STATUS_PROCESSING}).Find(&tasks).Error; dbErr != nil {
		err = fmt.Errorf("%s获取待处理和处理中任务数据异常！%v", errPrefix, dbErr)
		return
	}
	if len(tasks) == 0 {
		return
	}
	// 直接修改数据库
	for _, task := range tasks {
		if toCancelledErr := task.ToCancelled(reason); toCancelledErr != nil {
			global.Logger.WithField("strack", util.FormatError(toCancelledErr)).Errorf("%s任务 %d %s 取消执行失败！%v", errPrefix, task.Id, task.Title, toCancelledErr)
			continue
		}
		// 不管锁是否存在，都清理一下
		redis.NewLock(fmt.Sprintf("admin.task:%d", task.Id), 3).Clear()
	}
	return
}

// 立即执行
func ImmediateHandle(taskIds []uint64) (err error) {
	for _, taskId := range taskIds {
		go HandleTask(taskId)
	}
	return
}

// 重新执行
func ReHandle(taskIds []uint64) (err error) {
	if len(taskIds) == 0 {
		return
	}
	errPrefix := "任务重新执行："
	// 获取有效的锁和任务id
	locks := map[uint64]*redis.Lock{}
	reHandleTaskIds := make([]uint64, 0, len(taskIds))
	for _, taskId := range taskIds {
		lock := redis.NewLock(fmt.Sprintf("admin.task:%d", taskId), 300)
		ok, lockErr := lock.Get()
		if lockErr != nil {
			global.Logger.WithField("strack", util.FormatError(lockErr)).Errorf("%s任务 %d 锁获取异常！%v", errPrefix, taskId, lockErr)
			continue
		}
		if !ok {
			continue
		}
		locks[taskId] = lock
		reHandleTaskIds = append(reHandleTaskIds, taskId)
	}
	defer func() {
		for _, lock := range locks {
			lock.Release()
		}
	}()
	if len(reHandleTaskIds) == 0 {
		return
	}
	// 获取失败和已取消任务数据
	var tasks []*model.AdminTask
	if dbErr := global.MySQL.Where("id in ? and status in ?", reHandleTaskIds, []any{taskenum.STATUS_FAILED, taskenum.STATUS_CANCELLED}).Find(&tasks).Error; dbErr != nil {
		err = fmt.Errorf("%s获取失败和已取消任务数据异常！%v", errPrefix, dbErr)
		return
	}
	if len(tasks) == 0 {
		return
	}
	// 将任务数据状态修改为等待中
	handledTaskIds := make([]uint64, 0, len(tasks))
	for _, task := range tasks {
		if toPendingErr := task.ToPending("重新执行"); toPendingErr != nil {
			global.Logger.WithField("strack", util.FormatError(toPendingErr)).Errorf("%s任务 %d %s 锁获取异常！%v", errPrefix, task.Id, task.Title, toPendingErr)
			continue
		}
		handledTaskIds = append(handledTaskIds, task.Id)
	}
	// 执行任务
	for _, taskId := range handledTaskIds {
		// 先释放锁
		if lock, exists := locks[taskId]; exists {
			lock.Release()
			delete(locks, taskId)
		}
		go HandleTask(taskId)
	}
	return
}

// 处理单个任务
func HandleTask(id uint64) {
	errPrefix := fmt.Sprintf("处理任务 %d ", id)
	// 获取锁
	lock := redis.NewLock(fmt.Sprintf("admin.task:%d", id), 300)
	defer lock.Release()
	ok, err := lock.Get()
	if err != nil {
		global.Logger.WithField("strack", util.FormatError(err)).Errorf("%s锁获取异常！%v", errPrefix, err)
	}
	if !ok {
		return
	}
	var task *model.AdminTask
	if dbErr := global.MySQL.Where("id = ?", id).First(&task).Error; dbErr != nil && dbErr != gorm.ErrRecordNotFound {
		global.Logger.WithField("strack", util.FormatError(err)).Errorf("%s数据查询异常！%v", errPrefix, dbErr)
		return
	}
	if task == nil || task.Id == 0 {
		global.Logger.WithField("strack", util.FormatError(err)).Errorf("%s数据不存在！", errPrefix)
		return
	}
	if task.Status != taskenum.STATUS_PENDING {
		global.Logger.WithField("strack", util.FormatError(err)).Errorf("%s任务状态值需为等待中=%d才可被执行，当前值为：为：%d", errPrefix, taskenum.STATUS_PENDING, task.Status)
		return
	}
	// 根据任务类型获取处理者
	var handler TaskHandlerInterface
	switch task.Type {
	case taskenum.TYPE_TABLE_BUILDER_EXPORT:
		handler = NewTableBuilderExport()
	default:
		task.ToCancelled("当前任务类型暂不支持处理！")
		return
	}
	// 执行中状态
	if err := task.ToProcessing(0); err != nil {
		err = fmt.Errorf("变更处理中状态失败，原因：%v", err)
		task.ToFailed(err)
		global.Logger.WithField("strack", util.FormatError(err)).Errorf("%s%v", errPrefix, err)
		return
	}
	// 启动心跳，每5s更新心跳
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	go func() {
		ticker := time.NewTicker(consts.ADMIN_TASK_HEARTBEAT_TIME * time.Second)
		defer ticker.Stop()
		for {
			select {
			case <-ctx.Done():
				fmt.Printf("%s停止心跳", errPrefix)
				return
			case <-ticker.C:
				if err := task.Heartbeat(); err != nil {
					err = fmt.Errorf("心跳异常！%v", err)
					task.ToFailed(err)
					global.Logger.WithField("strack", util.FormatError(err)).Errorf("%s%v", errPrefix, err)
					return
				}
			}
		}
	}()
	// 执行任务
	handler.SetTask(task)
	if err := handler.Handle(); err != nil {
		err = fmt.Errorf("执行失败，原因：%v", err)
		task.ToFailed(err)
		global.Logger.WithField("strack", util.FormatError(err)).Errorf("%s%v", errPrefix, err)
	}
}

// 获取待处理任务
func getPendingTasks(taskIds ...uint64) (tasks []*model.AdminTask, err error) {
	tasksQuery := global.MySQL.Where("status = ?", taskenum.STATUS_PENDING).Order("created_at ASC")
	if len(taskIds) > 0 {
		// 有指定任务id
		tasksQuery.Where("id in ?", taskIds)
	} else {
		// 没有指定任务
		// 获取正在执行中的任务数量
		var processingCount int64
		if err = global.MySQL.Model(&model.AdminTask{}).Where("status = ?", taskenum.STATUS_PROCESSING).Count(&processingCount).Error; err != nil {
			err = fmt.Errorf("获取正在执行中的任务数量失败！%v", err)
			return
		}
		// 获取可执行的任务数量
		limit := consts.ADMIN_TASK_PROCESSING_MAX_COUNT - processingCount
		if limit <= 0 {
			return
		}
		tasksQuery.Limit(int(limit))
	}
	// 获取任务
	if err = tasksQuery.Find(&tasks).Error; err != nil {
		err = fmt.Errorf("获取等待中的后台任务数据异常！%v", err)
	}
	return
}

// 处理心跳停止任务
func HandleDeartbeatDie() (err error) {
	lock := redis.NewLock("admin.task.die", 300)
	defer lock.Release()
	ok, err := lock.Get()
	if err != nil {
		err = fmt.Errorf("后台任务心跳停止处理锁获取失败！%v", err)
		return
	}
	if !ok {
		return
	}
	currTime := time.Now()
	// 正在处理中的任务心跳时间超过4次未更新，则默认为程序处理被中断
	var tasks []*model.AdminTask
	if err = global.MySQL.Where("status = ? and updated_at < ?", taskenum.STATUS_PROCESSING, currTime.Add(-time.Duration(HeartbeatTimeoutTime())*time.Second)).Find(&tasks).Error; err != nil {
		err = fmt.Errorf("获取正在处理中的心跳停止的任务数据异常！%v", err)
		return
	}
	for _, task := range tasks {
		task.ToFailed("程序处理被中断")
	}
	return
}

// 心跳超时时间 = 4次心跳时间
func HeartbeatTimeoutTime() int {
	return consts.ADMIN_TASK_HEARTBEAT_TIME * 4
}

// 处理任务完成
func HandleComplated(task *model.AdminTask) (err error) {
	err = task.ToCompleted()
	if err == nil {
		// 创建任务完成通知
		err = noticeservice.Create(noticeenum.TYPE_TASK, task.MemberId, "任务完成："+task.Title, task.Id)
	}
	return
}
